#include "DSTH_IOCP.h"
#include <assert.h>
#include "IOCPOperation.h"
#include "Channel.h"
#include "Log.h"

namespace dsth {

DSTHIocp::DSTHIocp() {
    //TODO: set concurrency
    InitIocp(0);
}

DSTHIocp::~DSTHIocp() {
    if (iocp_handle_) ::CloseHandle(iocp_handle_);
}

void DSTHIocp::InitIocp(int concurrency) {
    iocp_handle_ = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0,
        static_cast<DWORD>((std::min<size_t>)(concurrency, DWORD(~0))));
    if (!iocp_handle_) {
        LOG_FATAL("Failed to create new iocp port. errorcode:%d", ::GetLastError());
    }
}

Timestamp DSTHIocp::Run(int time_out) {
    DWORD bytes_transferred = 0;
    ULONG_PTR completion_key = 0;
    LPOVERLAPPED overlapped = 0;
    ::SetLastError(0);
    BOOL ok = ::GetQueuedCompletionStatus(iocp_handle_, &bytes_transferred,
        &completion_key, &overlapped, time_out);
    DWORD last_error = ::GetLastError();
    Timestamp now_time(Timestamp::GetNowTime());

    if (overlapped) {
        IOCPOperation* op = static_cast<IOCPOperation*>(overlapped);
        int errcode = last_error;
        if (completion_key == kOverlappedContainsResult) {
            errcode = static_cast<int>(op->Offset);
            bytes_transferred = op->OffsetHigh;
        }
        else {            
            op->Offset = last_error;
            op->OffsetHigh = bytes_transferred;
        }

        if (::InterlockedCompareExchange(&op->ready_, 1, 0) == 1) {
            op->Complete(last_error, bytes_transferred);
        }
    }
    else if (!ok) {
        if (last_error != WAIT_TIMEOUT) {
            printf("error:%d", last_error);    
        }
        printf("time out\n");
    }
    else if (completion_key == kWakeForDispatch) {
        // We have been woken up to try to acquire responsibility for dispatching
        // timers and completed operations.
        printf("kWakeForDispatch\n");
    }
    else {
        //TODO: check stop
        printf("check stop\n");
    }
    return now_time;
}

void DSTHIocp::RegisterHandleOrDie(HANDLE handle) {
    if (::CreateIoCompletionPort(handle, iocp_handle_, 0, 0) == 0) {
        LOG_FATAL("Failed to bind socket to iocp port. errorcode:%d", ::GetLastError());
    }
}

void DSTHIocp::OnCompletion(IOCPOperation* op, DWORD errcode, DWORD bytes_transferred) {
    op->ready_ = 1;

    //op->Internal =; TODO: set me
    op->Offset = errcode;
    op->OffsetHigh = bytes_transferred;

    // Enqueue the operation on the I/O completion port.
    if (!::PostQueuedCompletionStatus(iocp_handle_,
        0, kOverlappedContainsResult, op)) {
        // Out of resources. Put on completed queue instead.
        //mutex::scoped_lock lock(dispatch_mutex_);
        //completed_ops_.push(op);
        //::InterlockedExchange(&dispatch_required_, 1);
    }
}

void DSTHIocp::OnPending(IOCPOperation* op) {
    if (::InterlockedCompareExchange(&op->ready_, 1, 0) == 1) {
        // Enqueue the operation on the I/O completion port.
        if (!::PostQueuedCompletionStatus(iocp_handle_,
            0, kOverlappedContainsResult, op)) {
            // Out of resources. Put on completed queue instead.
            //mutex::scoped_lock lock(dispatch_mutex_);
            //completed_ops_.push(op);
            //::InterlockedExchange(&dispatch_required_, 1);
        }
    }
}

} // namespace
